package com.saga.energy.sink;

import com.alibaba.druid.pool.DruidDataSource;
import com.saga.energy.bean.EnergyHistory;
import com.saga.energy.util.DruidDataSourceUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.Statement;
import java.util.List;

/**
 * @author wdl
 * @date 2022/5/5 18:48
 */
public class Energy_15_his_list_sink extends RichSinkFunction<List<EnergyHistory>> {


    private Integer recordCount = 0;
//    private DruidDataSource dataSource = null;

    private Connection connection = null;
    private PreparedStatement ps = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("初始化 DataSource ... ");
//            dataSource = DruidDataSourceUtil.getInstance();

        System.out.println("init mysql config ... ");
        String url = "jdbc:mysql://hadoop01:3306/saga_dev?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false";
        connection = DriverManager.getConnection(url,"root","j5ry0#jZ7vaUt5f4");
        String insertSql = "insert into energy_15_min_his(building, func_id, meter, data_time, data_value, dt) values (?, ?, ?, ?, ?, ?)";
        ps = connection.prepareStatement(insertSql);
    }

    @Override
    public void invoke(List<EnergyHistory> value, Context context) throws Exception {
        if (value.size() > 0){
            for (EnergyHistory energyHistory : value) {
                recordCount++;
                ps.setString(1, energyHistory.getBuilding());
                ps.setString(2, energyHistory.getFuncid());
                ps.setString(3, energyHistory.getMeter());
                ps.setString(4, energyHistory.getData_time());
                ps.setBigDecimal(5, energyHistory.getData_value());
                ps.setString(6, energyHistory.getDt());
                ps.addBatch();
            }

            System.out.println(" ---------------------------------- " + recordCount);
            int[] count = ps.executeBatch();
            System.out.println("******************** 插入 " + count.length + " 条数据！********************");
            recordCount = 0;
            if (connection != null) {
                connection.close();
            }
            if (ps != null) {
                ps.close();
            }
        }
    }

    @Override
    public void close() throws Exception {
        if (connection != null) {
            connection.close();
        }
        if (ps != null) {
            ps.close();
        }
    }

    private String getSql() {
        String sql_template = "insert into energy_15_min_his(building, func_id, meter, data_time, data_value, dt) values (?, ?, ?, ?, ?, ?)";
        return sql_template;
    }
}
